package com.qf.topic;

import com.qf.util.MQConnUtil;
import com.rabbitmq.client.*;
import org.junit.Test;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

import static com.qf.topic.Publisher.*;
// static 修饰    5个   成员变量，方法，类，代码块，导包

public class Consumer {

    @Test
    public void consume1() throws Exception, TimeoutException {
        //1. c
        Connection conn = MQConnUtil.getConn();

        //2. c
        Channel channel = conn.createChannel();

        //3. exchange
        channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.TOPIC);
        //4. queue
        channel.queueDeclare(QUEUE_NAME1,true,false,false,null);
        channel.queueDeclare(QUEUE_NAME2,true,false,false,null);
        //5. binding
        channel.queueBind(QUEUE_NAME1,EXCHANGE_NAME,ROUTING_KEY1);
        channel.queueBind(QUEUE_NAME2,EXCHANGE_NAME,ROUTING_KEY2);
        channel.queueBind(QUEUE_NAME2,EXCHANGE_NAME,ROUTING_KEY3);


        //3. 监听queue
        DefaultConsumer callback = new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("消费者监听" + Publisher.QUEUE_NAME1 + "队列,获取到消息:" + new String(body));
                channel.basicAck(envelope.getDeliveryTag(),false);
            }
        };
        channel.basicConsume(Publisher.QUEUE_NAME1,callback);

        //4. stop
        System.out.println(Publisher.QUEUE_NAME1 + "开始监听");
        System.in.read();

        //5. c
        channel.close();
        conn.close();


    }


    @Test
    public void consume2() throws Exception, TimeoutException {
//1. c
        Connection conn = MQConnUtil.getConn();

        //2. c
        Channel channel = conn.createChannel();

        //3. exchange
        channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.TOPIC);
        //4. queue
        channel.queueDeclare(QUEUE_NAME1,true,false,false,null);
        channel.queueDeclare(QUEUE_NAME2,true,false,false,null);
        //5. binding
        channel.queueBind(QUEUE_NAME1,EXCHANGE_NAME,ROUTING_KEY1);
        channel.queueBind(QUEUE_NAME2,EXCHANGE_NAME,ROUTING_KEY2);
        channel.queueBind(QUEUE_NAME2,EXCHANGE_NAME,ROUTING_KEY3);


        //3. 监听queue
        DefaultConsumer callback = new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("消费者监听" + Publisher.QUEUE_NAME2 + "队列,获取到消息:" + new String(body));
                channel.basicAck(envelope.getDeliveryTag(),false);
            }
        };
        channel.basicConsume(Publisher.QUEUE_NAME2,callback);

        //4. stop
        System.out.println(Publisher.QUEUE_NAME2 + "开始监听");
        System.in.read();

        //5. c
        channel.close();
        conn.close();


    }
}
